package cn.mesmile.flink.sink;

import cn.mesmile.flink.jdkstream.VideoOrder;
import cn.mesmile.flink.source.VideoOrderSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zb
 * @date 2021/8/22 0:04
 * @Description
 */
public class FlinkCustomSinkMysqlApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 自动模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);

        VideoOrderSource videoOrderSource = new VideoOrderSource();
        DataStreamSource<VideoOrder> videoOrderDataStreamSource = env.addSource(videoOrderSource);
        // 过滤
        SingleOutputStreamOperator<VideoOrder> filter = videoOrderDataStreamSource.filter(new FilterFunction<VideoOrder>() {
            @Override
            public boolean filter(VideoOrder videoOrder) throws Exception {
                return videoOrder.getMoney() > 50;
            }
        }).setParallelism(2);

        MysqlSink mysqlSink = new MysqlSink();
        // 设置输出
        filter.addSink(mysqlSink);
        filter.print("result_").setParallelism(3);
        // 执行
        env.execute("mysql job");


    }
}
